/*
 * simplemq.hpp
 *
 *  Created on: 2020年4月5日
 *      Author: guyadong
 */

#ifndef SIMPLEMQ_SIMPLEMQ_HPP_
#define SIMPLEMQ_SIMPLEMQ_HPP_
#include <exception>
#include <string>
#include <memory>
#include <algorithm>
#include <functional>
#include <vector>
#include <map>
#include <set>
#include <unordered_set>
#include <iterator>
#include <initializer_list>
#include <sstream>
#include <ThreadPool.hpp>
#include <RWLock.hpp>
#include <json_utilits.hpp>
#include <sample_log.h>
#include <tsl/ordered_set.h>
#include "simplemq_commons.hpp"

#define SIMPLEMQ_CPP_VERSION_MAJOR 0
#define SIMPLEMQ_CPP_VERSION_MINOR 0
#define SIMPLEMQ_CPP_VERSION_PATCH 2
#define SIMPLEMQ_CPP_VERSION_SNAPSHOT 1 
#define SIMPLEMQ_CPP_VERSION ((SIMPLEMQ_CPP_VERSION_MAJOR<<16)|(SIMPLEMQ_CPP_VERSION_MINOR << 8) |SIMPLEMQ_CPP_VERSION_PATCH)

namespace std {
#ifndef _HASH_EQUAL_LESS_OF_FUNCTION_
#define _HASH_EQUAL_LESS_OF_FUNCTION_
	/** 为std::function提供std::hash特例化实现 */
	template< class R, class... Args >
	struct hash<std::function<R(Args...)>> {
		typedef std::function<R(Args...)> argument_type;
		typedef std::size_t result_type;
		result_type operator()(argument_type _f) const noexcept {
			return (result_type)(_f.template target<R(Args...)>());
		}
	};
	template<class R, class... Args>
	struct equal_to<std::function<R(Args...)>>
	{	// functor for operator==
		typedef std::function<R(Args...)> first_argument_type;
		typedef std::function<R(Args...)> second_argument_type;
		typedef bool result_type;

		constexpr bool operator()(first_argument_type _Left, second_argument_type _Right) const
		{	// apply operator== to operands
			return (_Left.template target<R(Args...)>() == _Right.template target<R(Args...)>());
		}
	};
	template<class R, class... Args>
	struct less<std::function<R(Args...)>>
	{	// functor for operator<
		typedef std::function<R(Args...)> first_argument_type;
		typedef std::function<R(Args...)> second_argument_type;
		typedef bool result_type;

		constexpr bool operator()(const first_argument_type& _Left, const second_argument_type& _Right) const
		{	// apply operator< to operands
			return (_Left.template target<R(Args...)>() < _Right.template target<R(Args...)>());
		}
	};
#endif /** _HASH_EQUAL_LESS_OF_FUNCTION_ */
}
namespace simplemq {
	using json = nlohmann::ordered_json;
	/**
	 * 取消订阅异常, {@link IMessageAdapter}接口抛出该异常时,{@link IMessageDispatcher}对象会取消订阅当前频道<br>
	 * 参见 {@link BaseSubscriber#dispatch(String, String)} 实现代码
	 */
	class SmqUnsubscribeException : std::exception {
	public:
		const std::string message;
		/** 是否注销频道 */
		const bool unregister;
	public:
		/**
		 * 构造函数
		 * @param message 异常信息
		 * @param unregister 是否注销频道
		 */
		SmqUnsubscribeException(const std::string &message  = "", bool unregister = false) : message(message), unregister(unregister) {}
		const char* what() const throw() {
			return message.c_str();
		}
		SmqUnsubscribeException(bool unregister):SmqUnsubscribeException("",unregister){}
	};
	/** 订阅消息数据处理接口(Function) */
	typedef std::function<void(const simplemq::json &)> message_adapter_f;
	struct IMessageAdapter;
	using message_adapter_ptr = std::shared_ptr<IMessageAdapter>;
	/** 订阅消息数据处理接口(Class) */
	struct IMessageAdapter
	{
		virtual ~IMessageAdapter() = default;
		/**
		 * 订阅消息处理
		 * @param t 消息对象
		 * @throws SmqUnsubscribeException 取消订阅当前频道异常指示
		 */
		virtual void onSubscribe(const simplemq::json &t) = 0;
		/**
		 * 订阅消息处理
		 * @param t 消息对象
		 * @throws SmqUnsubscribeException 取消订阅当前频道异常指示
		 */
		virtual void onSubscribe(const std::string &t) {
			try
			{
				auto j = simplemq::json::parse(t);
				onSubscribe(j);
			}
			catch (/** 可能抛出的json解析异常 */simplemq::json::parse_error & e) {
				SAMPLE_ERR("ERROR {}: {}", typeid(e).name(), e.what());
			}
			catch (/** 可能抛出的其他json异常 */simplemq::json::exception & e) {
				SAMPLE_ERR("ERROR {}: {}", typeid(e).name(), e.what());
			}
		}
	};
	class IMessageAdapterFunImpl : public virtual IMessageAdapter
	{
		message_adapter_f adapter_f;
	public:
		IMessageAdapterFunImpl(const message_adapter_f &adapter_f = nullptr) 
			:adapter_f(adapter_f ? adapter_f : [](const simplemq::json & j) {}/** 为null提供空函数实现 */) {}
		virtual ~IMessageAdapterFunImpl() = default;
		/**
		 * 订阅消息处理
		 * @param t 消息对象
		 * @throws SmqUnsubscribeException 取消订阅当前频道异常指示
		 */
		virtual void onSubscribe(const simplemq::json &t) override
		{
			adapter_f(t);
		}
		/**
		 * 订阅消息处理
		 * @param t 消息对象
		 * @throws SmqUnsubscribeException 取消订阅当前频道异常指示
		 */
		virtual void onSubscribe(const std::string &t) override {
			try
			{
				auto j = simplemq::json::parse(t);
				onSubscribe(j);
			}
			catch (/** 可能抛出的json解析异常 */simplemq::json::parse_error & e) {
				SAMPLE_ERR("ERROR {}: {}", typeid(e).name(), e.what());
			}
			catch (/** 可能抛出的其他json异常 */simplemq::json::exception & e) {
				SAMPLE_ERR("ERROR {}: {}", typeid(e).name(), e.what());
			}
		}
		// == operator
		bool operator==(const IMessageAdapterFunImpl & rhs)const {
			return adapter_f.target<void(*)(const simplemq::json &)>() == rhs.adapter_f.target<void(*)(const simplemq::json &)>();
		}
	};
	class Channel;
	/**  频道注销侦听器(Function) */
	using unregister_listener_fun = std::function<void(const Channel &)>;

	/**
	 * (消息)频道对象定义<br>
	 * 这里的频道概念可以是发布/订阅模型中的频道,也可以是生产者/消费者模型中的队列
	 *
	 */
	class Channel {
	public:
		/**  频道名(消息来源) */
		std::string name;
		/**  频道对应的消息处理器 */
		message_adapter_ptr adapter;
	private:
		tsl::ordered_set<unregister_listener_fun> unregistedListeners;
	public:
		/**
		 * 构造函数
		 * @param name 频道名
		 * @param handle 消息处理函数
		 */
		Channel(const std::string &name, const message_adapter_ptr &adapter) : name(name), adapter(adapter){}
		/**
		 * 构造函数
		 * @param name 频道名
		 * @param handle 消息处理函数
		 */
		Channel(const std::string &name, const message_adapter_f &handle) 
			: name(name), adapter(std::make_shared<IMessageAdapterFunImpl>(handle)) {}
		/**
		 * 构造函数
		 * @param name 频道名
		 */
		Channel(const std::string &name) : name(name) {
		}
		Channel() = default;

		Channel(const Channel &) = default;
		Channel(Channel &&) = default;

		~Channel() = default;

		// = operator
		Channel& operator=(const Channel&) = default;
		// < operator 
		bool operator<(const Channel & rhs) const {
			return name.compare(rhs.name) < 0;
		}
		// == operator
		bool operator==(const Channel & rhs)const {
			return name == rhs.name && adapter == rhs.adapter;
		}

		std::string to_string() const
		{
			std::stringstream ss;
			ss << "Channel(name=" << name << ")";
			return ss.str();
		}

		static std::vector<std::string> namesFrom(const std::map<std::string, Channel> &m)
		{
			std::vector<std::string> names;
			std::transform(begin(m), end(m), std::back_inserter(names),
				[](const std::decay<decltype(m)>::type::value_type & pair) {
				return pair.first;
			});
			return names;
		}
		template<typename InputIt = std::iterator<std::input_iterator_tag, Channel>>
		static std::vector<std::string> namesFrom(InputIt first, InputIt last)
		{
			std::vector<std::string> names;
			std::transform(first, last, std::back_inserter(names),
				[](const Channel & v) {
				return v.name;
			});
			return names;
		}
		/**
		 * 当前频道被注销时调用
		 */
		void onUnregisted() {
			for (auto itor = unregistedListeners.begin(); itor != unregistedListeners.end();)
			{
				(*itor)(*this);
				itor = unregistedListeners.erase(itor);
			}
		}
		/**
		 * 添加注销侦听器
		 * @param unregistedListener
		 * @return
		 */
		Channel& addUnregistedListener(const unregister_listener_fun & unregistedListener) {
			throw_if_null(unregistedListener);
			this->unregistedListeners.emplace(unregistedListener);
			return *this;
		}
	};
	/**
	 * 消息分发接口,用于将收到的消息分派到不同的消息处理器
	 */
	class IMessageDispatcher {
	public:
		/**
		 * @brief 消息分发
		 * @param channel 收到消息的频道名
		 * @param message 消息内容
		 */
		virtual void dispatch(const std::string &channel, const std::string &message) = 0;
		virtual ~IMessageDispatcher() = default;
	};
	/**
	 * (消息)频道注册订阅接口
	 */
	class IMessageRegister {
	public:
		/**
		 * 订阅频道
		 * @param channels 频道名列表,为null或空时订阅所有 {@link #channelSubs}中的频道
		 * @return 返回实际订阅的频道列表
		 */
		virtual std::set<std::string> subscribe(std::initializer_list<std::string> channels) = 0;
		/**
		 * 取消频道订阅
		 * @param channels 频道名列表,为null或空时取消订阅所有 {@link #channelSubs}中的频道
		 * @return 返回实际取消订阅的频道列表
		 */
		virtual std::set<std::string> unsubscribe(std::initializer_list<std::string> channels) = 0;
		/**
		 * 注册并订阅指定的频道
		 * @param channels
		 * @return 返回实际注册的频道名列表
		 */
		virtual std::set<Channel> _register(std::initializer_list<Channel> channels) = 0;
		/**
		 * 取消订阅指定的频道,并注销频道
		 * @param channels
		 * @return 返回实际注销的频道名列表
		 */
		virtual std::set<std::string> unregister(std::initializer_list<Channel> channels) = 0;
		/**
		 * 取消订阅指定的频道,并注销频道
		 * @param channels
		 * @return 返回实际注销的频道名列表
		 */
		virtual std::set<std::string> unregister(std::initializer_list<std::string> channels) = 0;
		/**
		 * 返回注册的 {@link Channel}对象
		 * @param channel 频道名
		 * @return
		 */
		virtual std::shared_ptr<Channel> getChannel(const std::string &channel) = 0;
		virtual ~IMessageRegister() = default;
	};
	/**
	 * 消息订阅接口
	 */
	class ISubscriber : public IMessageDispatcher, public IMessageRegister {
	public:
		virtual ~ISubscriber() = default;
	};
	/**
	 * 消息订阅者抽象类
	 *
	 */
	class BaseSubscriber : public ISubscriber {
	private:
		gdface::mt::RWLock rwlock;
		/** 用于执行消息处理的单线程池 */
		::gdface::ThreadPool subPool;
	protected:
		std::map<std::string, Channel> channelRegs;
		/** 订阅的频道名 */
		std::set<std::string> subChannelSet;
		/**
		 * 订阅频道具体实现
		 * @param channels 频道名列表,不可为null或空
		 */
		virtual void doSubscribe(const std::vector<std::string> &channels) = 0;
		/**
		 * 取消订阅具体实现
		 * @param channels 频道名列表,为null或空时订阅所有 {@link #channelSubs}中的频道
		 */
		virtual void doUnsubscribe(const std::vector<std::string> &channels) = 0;
		/** 启动实例 */
		virtual void open() = 0;
	public:
		BaseSubscriber() : channelRegs(), subChannelSet(), subPool(1){};
		~BaseSubscriber() = default;
		std::set<Channel> _register(const std::vector<Channel> &channels)
		{
			auto guard = rwlock.write_guard();
			// 新注册的频道
			std::set<Channel> chs;
			for (auto &ch : channels) {
				if (channelRegs.count(ch.name) == 0) {
					channelRegs[ch.name] = ch;
					chs.emplace(ch);
				}
			}
			auto names = Channel::namesFrom(chs.begin(), chs.end());
			subscribe(names);
			return chs;
		}
		std::set<std::string> unregister(const std::vector<std::string> &channels)
		{
			auto guard = rwlock.write_guard();
			std::set<std::string> erased;
			if (!channels.empty()) {
				unsubscribe(channels);
				for (auto name : channels) {
					if (channelRegs.count(name)) {
						auto &ch = channelRegs[name];
						ch.onUnregisted();
						erased.emplace(name);
						channelRegs.erase(name);
					}
				}
			}
			return erased;
		}
		std::set<std::string> unregister(const std::vector<Channel> &channels) {
			return unregister(Channel::namesFrom(channels.begin(), channels.end()));
		}
		std::set<std::string> subscribe(const std::vector<std::string>&channels)
		{
			auto guard = rwlock.write_guard();
			auto chs = channels.empty() ? Channel::namesFrom(channelRegs) : channels;
			std::set<std::string> diff;
			// diff = chs - subChannelSet
			std::set_difference(chs.begin(), chs.end(), subChannelSet.begin(), subChannelSet.end(), std::inserter(diff, diff.end()));
			subChannelSet.insert(diff.begin(), diff.end());
			doSubscribe(std::vector<std::string>(diff.begin(), diff.end()));
			return diff;
		}
		std::set<std::string> unsubscribe(const std::vector<std::string>&channels)
		{
			auto guard = rwlock.write_guard();
			auto chs = channels.empty() ? Channel::namesFrom(channelRegs) : channels;
			std::set<std::string> diff;
			// diff = subChannelSet - chs
			std::set_difference(subChannelSet.begin(), subChannelSet.end(), chs.begin(), chs.end(), std::inserter(diff, diff.end()));
			std::set<std::string> erased;
			// erased = subChannelSet AND chs 
			std::set_intersection(subChannelSet.begin(), subChannelSet.end(), chs.begin(), chs.end(), std::inserter(erased, erased.end()));
			subChannelSet = diff;
			doUnsubscribe(std::vector<std::string>(erased.begin(), erased.end()));
			return erased;
		}

		std::set<std::string> subscribe(std::initializer_list<std::string> channels)override {
			return subscribe(std::vector<std::string>(channels));
		}
		std::set<std::string> unsubscribe(std::initializer_list<std::string> channels)override {
			return unsubscribe(std::vector<std::string>(channels));
		}
		std::set<Channel> _register(std::initializer_list<Channel> channels)override {
			return _register(std::vector<Channel>(channels));
		}
		std::set<std::string> unregister(std::initializer_list<Channel> channels)override {
			return unregister(std::vector<Channel>(channels));
		}
		std::set<std::string> unregister(std::initializer_list<std::string> channels)override {
			return unregister(std::vector<std::string>(channels));
		}

		std::shared_ptr<Channel> getChannel(const std::string &channel) override
		{
			auto guard = rwlock.read_guard();
			return channelRegs.count(channel) != 0 ? std::make_shared<Channel>(channelRegs[channel]) : nullptr;
		}

		void dispatch(const std::string & channel, const std::string & message) override
		{
			if (channelRegs.count(channel) != 0)
			{
				auto &ch = channelRegs[channel];
				if (ch.adapter)
				{
					subPool.enqueue([&, channel, message]() {
						try
						{
							try
							{
								auto j = simplemq::json::parse(message);
								ch.adapter->onSubscribe(j);
							}
							catch (const std::exception&)
							{
								ch.adapter->onSubscribe(message);
							}
						}
						catch (const SmqUnsubscribeException& e)
						{
							if (e.unregister) {
								unregister({ channel });
							}
							else {
								unsubscribe({ channel });
							}
							SAMPLE_LOG("unsubscribe channel: {}", channel);
						}
						catch (const std::exception &e) {
							SAMPLE_ERR("EXCEPTION:{}: {}", typeid(e).name(), e.what());
						}
						catch (...) {
							SAMPLE_ERR("SOMETHING ERROR");
						}
					});
				}
			}
			else {
				SAMPLE_OUT("unregistered channel: '{}'", channel);
			}
		}

	};
	/**
	 * 基础消息发布接口
	 */
	struct publisher {
		/**
		 * 同步方式向指定频道发布消息
		 * @param channel 频道名
		 * @param message 消息
		 * @return 收到消息的client数量
		 */
		virtual int64_t publish(const std::string &channel, const std::string& message) = 0;
		~publisher() = default;
	};
	/**
	 * 结构化类型消息发布类
	 * @param <T> 消息类型
	 */
	class IPublisher {
	private:
		template<typename T>
		std::string to_json(const T&t) {
			simplemq::json j = t;
			return nlohmann::to_string(j);
		}
	public:
		IPublisher(const std::shared_ptr<publisher> &pub) : pub(pub) {}
		IPublisher() : IPublisher(nullptr) {};

		/**
		 * 向指定的频道发布消息
		 * @param channel 频道名
		 * @param object 消息对象
		 * @return the number of clients that received the message
		 */
		template<typename T>
		int64_t publish(const std::string &channel, const T& object) {
			if (pub) {
				pub->publish(channel, to_json(object));
			}
			return 0;
		}
		/**
		 * 向指定的频道发布一组消息
		 * @param channel 频道名
		 * @param first 起始迭代器
		 * @param last 结束迭代器
		 */
		template<typename T,class InputIt = std::iterator<std::input_iterator_tag, T>>
		void publish(const std::string &channel, InputIt first, InputIt last) {
			std::for_each(first, last, [&](const T &t) {
				publish(channel, t);
			});
		};
		/**
		 * 向指定的频道发布一组消息
		 * @param channel 频道名
		 * @param objects 消息对象列表
		 */
		template<typename T>
		void publish(Channel channel, std::initializer_list<T> objects) {
			publish(channel, objects.begin(), objects.end());
		};

	private:
		std::shared_ptr<publisher> pub;

	};
	using subscriber_ptr = std::shared_ptr<ISubscriber>;
	using publisher_ptr = std::shared_ptr<IPublisher>;
	struct BaseClientManager {
		virtual ~BaseClientManager() = default;
		/**
		 * 返回host,port指定的ISubscriber实例引用,
		 * @param params parameters to be connected to
		 */
		virtual subscriber_ptr getSubscriber(const base_client_params_ptr& params) = 0;
		/**
		 * 返回访问uri指定的ISubscriber实例引用,
		 * 对于每一个不同的uri,返回对应的单实例
		 * @param uri uri for connect  server
		 */
		virtual subscriber_ptr getSubscriber(const uri &uri) = 0;
		/**
		 * 返回访问uri指定的mqtt服务的MqttSubscriber实例引用,
		 * 对于每一个不同的uri,返回对应的单实例
		 * @param params parameters for connect mqtt server
		 */
		virtual subscriber_ptr  getSubscriber(const simplemq::json &params) = 0;
		/**
		 * 创建 IPublisher 消息发布实例
		 * @param params parameters for connect server
		 */
		virtual publisher_ptr getPublisher(const base_client_params_ptr & params) = 0;
		/**
		 * 创建 IPublisher 消息发布实例
		 * @param uri uri for connect server
		 */
		virtual publisher_ptr getPublisher(const uri & uri) = 0;
		/**
		* 创建 IPublisher 消息发布实例
		* @param params parameters for connect mqtt server
		* @return IPublisher 实例
		*/
		virtual publisher_ptr getPublisher(const simplemq::json & params) = 0;
		/** 返回实现类型: MQTT/REDIS */
		virtual std::string getImplType()const = 0;
		/** 将JSON转为base_mq_client_params */
		virtual base_client_params_ptr asClientParam(const simplemq::json & params) = 0;
	};
	using base_cmgr_ptr = std::shared_ptr<BaseClientManager>;
}/** namespace simplemq */
							
/* 为Channel提供std::hash<Channel>特例化实现 */
namespace std
{
	template<>
	struct hash<simplemq::Channel> {
		typedef simplemq::Channel argument_type;
		typedef std::size_t result_type;
		result_type operator()(argument_type const& ch) const noexcept {
			return std::hash<std::string>()(ch.to_string());
		}
	};
} /* namespace std */
#endif /* SIMPLEMQ_SIMPLEMQ_HPP_ */
